package com.gzc.just.play.last.war.common.share.distributed;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Supplier;

/**
 * Distributed lock
 * Based on Redis implementation
 */
@Component
@ConditionalOnClass(RedisConnectionFactory.class)
@ConditionalOnProperty(name = "allinone.enable-redis", havingValue = "true", matchIfMissing = true)
public class DistributedLock {

    private static final Logger log = LoggerFactory.getLogger(DistributedLock.class);

    private final RedisTemplate<String, Object> redisTemplate;
    
    // Local lock to reduce Redis access
    private final ReentrantLock localLock = new ReentrantLock();
    
    // Lock prefix
    private static final String LOCK_PREFIX = "distributed_lock:";
    
    // Default lock timeout in seconds
    private static final long DEFAULT_LOCK_TIMEOUT = 30L; // seconds
    
    @Autowired(required = false)
    public DistributedLock(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
    
    /**
     * Acquire lock
     *
     * @param lockKey Lock key
     * @return Lock value, used for verification when releasing lock
     */
    public String lock(String lockKey) {
        return lock(lockKey, DEFAULT_LOCK_TIMEOUT);
    }
    
    /**
     * Acquire lock
     *
     * @param lockKey Lock key
     * @param timeout Lock timeout (seconds)
     * @return Lock value, used for verification when releasing lock
     */
    public String lock(String lockKey, long timeout) {
        String lockValue = UUID.randomUUID().toString();
        String redisKey = LOCK_PREFIX + lockKey;
        
        // First try to acquire local lock
        localLock.lock();
        try {
            // Try to acquire distributed lock
            Boolean locked = redisTemplate.opsForValue().setIfAbsent(redisKey, lockValue, timeout, TimeUnit.SECONDS);
            
            if (Boolean.TRUE.equals(locked)) {
                log.debug("Acquired distributed lock: {}", lockKey);
                return lockValue;
            }
            
            log.debug("Failed to acquire distributed lock: {}", lockKey);
            return null;
        } finally {
            localLock.unlock();
        }
    }
    
    /**
     * Release lock
     *
     * @param lockKey Lock key
     * @param lockValue Lock value
     * @return Whether release was successful
     */
    public boolean unlock(String lockKey, String lockValue) {
        if (lockValue == null) {
            return false;
        }
        
        String redisKey = LOCK_PREFIX + lockKey;
        
        // First try to acquire local lock
        localLock.lock();
        try {
            // Lua script ensures atomicity: first compare lock value, then delete
            String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
                              "  return redis.call('del', KEYS[1]) " +
                              "else " +
                              "  return 0 " +
                              "end";
            
            Long result = redisTemplate.execute((RedisCallback<Long>) connection ->
                connection.eval(
                    luaScript.getBytes(),
                    ReturnType.INTEGER,
                    1,
                    redisKey.getBytes(),
                    lockValue.getBytes()
                )
            );
            
            boolean unlocked = result != null && result > 0;
            if (unlocked) {
                log.debug("Released distributed lock: {}", lockKey);
            } else {
                log.warn("Failed to release distributed lock: {} (value mismatch or expired)", lockKey);
            }
            
            return unlocked;
        } finally {
            localLock.unlock();
        }
    }
    
    /**
     * Try to acquire lock (non-blocking)
     *
     * @param lockKey Lock key
     * @return Whether acquisition was successful
     */
    public boolean tryLock(String lockKey) {
        return tryLock(lockKey, DEFAULT_LOCK_TIMEOUT);
    }
    
    /**
     * Try to acquire lock (non-blocking)
     *
     * @param lockKey Lock key
     * @param timeout Lock timeout (seconds)
     * @return Whether acquisition was successful
     */
    public boolean tryLock(String lockKey, long timeout) {
        String lockValue = lock(lockKey, timeout);
        return lockValue != null;
    }
    
    /**
     * Try to acquire lock (blocking for specified time)
     *
     * @param lockKey Lock key
     * @param waitTime Wait time (milliseconds)
     * @param timeout Lock timeout (seconds)
     * @return Whether acquisition was successful
     */
    public boolean tryLock(String lockKey, long waitTime, long timeout) {
        long endTime = System.currentTimeMillis() + waitTime;
        
        while (System.currentTimeMillis() < endTime) {
            if (tryLock(lockKey, timeout)) {
                return true;
            }
            
            try {
                Thread.sleep(100); // Short sleep to avoid high CPU usage
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return false;
            }
        }
        
        return false;
    }
    
    /**
     * Execute operation with lock
     *
     * @param lockKey Lock key
     * @param supplier Operation
     * @param <T> Return type
     * @return Operation result
     */
    public <T> T executeWithLock(String lockKey, Supplier<T> supplier) {
        return executeWithLock(lockKey, DEFAULT_LOCK_TIMEOUT, supplier);
    }
    
    /**
     * Execute operation with lock
     *
     * @param lockKey Lock key
     * @param timeout Lock timeout (seconds)
     * @param supplier Operation
     * @param <T> Return type
     * @return Operation result
     */
    public <T> T executeWithLock(String lockKey, long timeout, Supplier<T> supplier) {
        String lockValue = lock(lockKey, timeout);
        if (lockValue == null) {
            throw new RuntimeException("Failed to acquire lock: " + lockKey);
        }
        
        try {
            return supplier.get();
        } finally {
            unlock(lockKey, lockValue);
        }
    }
    
    /**
     * Check if lock exists
     *
     * @param lockKey Lock key
     * @return Whether lock exists
     */
    public boolean isLocked(String lockKey) {
        String redisKey = LOCK_PREFIX + lockKey;
        return Boolean.TRUE.equals(redisTemplate.hasKey(redisKey));
    }
    
    /**
     * Extend lock validity period
     *
     * @param lockKey Lock key
     * @param lockValue Lock value
     * @param timeout New timeout (seconds)
     * @return Whether extension was successful
     */
    public boolean renewLock(String lockKey, String lockValue, long timeout) {
        if (lockValue == null) {
            return false;
        }
        
        String redisKey = LOCK_PREFIX + lockKey;
        
        // Lua script ensures atomicity: first compare lock value, then extend validity
        String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
                          "  return redis.call('expire', KEYS[1], ARGV[2]) " +
                          "else " +
                          "  return 0 " +
                          "end";
        
        Long result = redisTemplate.execute((RedisCallback<Long>) connection ->
            connection.eval(
                luaScript.getBytes(),
                ReturnType.INTEGER,
                1,
                redisKey.getBytes(),
                lockValue.getBytes(),
                String.valueOf(timeout).getBytes()
            )
        );
        
        boolean renewed = result != null && result > 0;
        if (renewed) {
            log.debug("Renewed distributed lock: {} for {} seconds", lockKey, timeout);
        } else {
            log.warn("Failed to renew distributed lock: {} (value mismatch or expired)", lockKey);
        }
        
        return renewed;
    }
}